#include <chrono>
#include <ctime>
#include <fstream>
#include <pthread.h>
#include <queue>
#include <stdlib.h>
#include <time.h>
#include <string>
#include <string.h>  
#include <thread>

#include <arpa/inet.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <unistd.h>

#include "openssl/hmac.h"
#include "examples/webrtc-client/easywsclient.hpp"
#include "rtc_base/ssl_adapter.h"
#include "rtc_base/thread.h"
#include "p2p/base/stun_server.h"

#include "api/video_codecs/builtin_video_decoder_factory.h"
#include "api/video_codecs/builtin_video_encoder_factory.h"
#include "api/audio_codecs/builtin_audio_encoder_factory.h"
#include "api/audio_codecs/builtin_audio_decoder_factory.h"
#include "api/rtc_event_log/rtc_event_log_factory.h"
#include "api/task_queue/default_task_queue_factory.h"
#include "media/engine/webrtc_media_engine.h"
#include "modules/audio_device/include/fake_audio_device.h"

#include "json/json.h"

#include "examples/webrtc-client/websocket_client.h"

const char kStreamLabel[] = "stream_label";
const char kAudioLabel[] = "audio_label";
const char kVideoLabel[] = "video_label";
const char kCandidateSdpMidName[] = "sdpMid";
const char kCandidateSdpMlineIndexName[] = "sdpMLineIndex";
const char kCandidateSdpName[] = "candidate";
const char kSessionDescriptionTypeName[] = "type";
const char kSessionDescriptionSdpName[] = "sdp";

static easywsclient::WebSocket::pointer websocket_client_webrtc = NULL;
std::mutex websocket_client_webrtc_lock;
bool loginInSuccessWebrtc = false;
std::string host_address_webrtc;
std::string userNameInternalWebrtc;
std::string serverAddressInternalWebrtc;
rtc::scoped_refptr<webrtc::PeerConnectionFactoryInterface> peer_connection_factory_webrtc = NULL;
bool stopedWebrtc = true;
std::mutex stopedWebrtcLock;

std::mutex conductor_channel_used_webrtc_lock;

bool hasPeerWebrtc1 = false;
bool hasPeerWebrtc2 = false;
bool hasPeerWebrtc3 = false;



// PEER传来的消息或WEBRTC事件
typedef struct tagWebrtcEvent {
    int eventType;
    std::string peerName;
    std::string eventData;
} WebrtcEvent;

std::mutex queue_lock_webrtc;
std::queue<WebrtcEvent> webrtcEventQueue;

void pushEvent(int eventType, const std::string& peerName, const std::string& eventData) {
    WebrtcEvent e;
    e.eventType = eventType;
    e.peerName = peerName;
    e.eventData = eventData;
    std::lock_guard<std::mutex> lock_(queue_lock_webrtc, std::adopt_lock);
    webrtcEventQueue.push(e);
}

int popEvent(WebrtcEvent& e) {
    std::lock_guard<std::mutex> lock_(queue_lock_webrtc, std::adopt_lock);
    if(webrtcEventQueue.empty()) {
        return -1;
    } else {
        e = webrtcEventQueue.front();
        webrtcEventQueue.pop();
        return 0;
    }
}

void websocketSendMessage(const std::string& message) {
    std::lock_guard<std::mutex> lock_ws(websocket_client_webrtc_lock, std::adopt_lock);
    if(websocket_client_webrtc == NULL) {
        fprintf(stdout, "WebSocket closed, can not send message\n");
        return;
    }
    easywsclient::WebSocket::readyStateValues webSocketState;
    webSocketState = websocket_client_webrtc->getReadyState();
    if(webSocketState == easywsclient::WebSocket::OPEN) {
        //fprintf(stdout, "websocketSendMessage: %s\n", message.c_str());
        websocket_client_webrtc->send(message.c_str());
    } else {
        fprintf(stdout, "WebSocket not opened, can not send message\n");
    }
}

class Conductor {
    /* inner class implement SetSessionDescriptionObserver
       CreateSessionDescriptionObserver and PeerConnectionObserver
    */
    class DataChannelObserverImp : public webrtc::DataChannelObserver {
    private:
        Conductor* parent;
    public:
        DataChannelObserverImp(Conductor* parent) : parent(parent) {}

        void OnStateChange() override {
            if(parent->data_channel.get() == NULL) {
                fprintf(stdout, "data channel null\n");
                return;
            }
            if (parent->data_channel->state() == webrtc::DataChannelInterface::kOpen) {
                fprintf(stdout, "data channel open\n");
                if(parent->getDataChannelStatus()) {
                    pthread_t pthread_id;
                    pthread_create(&pthread_id, NULL, parent->sendData, (void* )(parent));
                    pthread_detach(pthread_id);
                }
            }
        };
        void OnMessage(const webrtc::DataBuffer& buffer) override {};
        void OnBufferedAmountChange(uint64_t previous_amount) override {};
    };
    
    class SetSessionDescriptionObserverImp : public webrtc::SetSessionDescriptionObserver {
    public:
        virtual void OnSuccess() {
        }
        virtual void OnFailure(const std::string& error) {
            fprintf(stderr, "Error on SetSessionDescriptionObserverImp:OnFailure\n");
        }
        int AddRef() const override {return 0;}
        int Release() const override {return 0;}
    };


    class CreateSessionDescriptionObserverImp : public webrtc::CreateSessionDescriptionObserver {
    private:
        Conductor& parent;
    public:
        //CreateSessionDescriptionObserverImp() {}
        CreateSessionDescriptionObserverImp(Conductor& parent) : parent(parent) {}
        // CreateSessionDescriptionObserver implementation.
        // Called on success of Create{Offer,Answer}().
        virtual void OnSuccess(webrtc::SessionDescriptionInterface* desc){
            parent.OnSuccessCallBack(desc);
        }
        virtual void OnFailure(const std::string& error) {
            fprintf(stderr, "Error on CreateSessionDescriptionObserverImp:OnFailure\n");
        }
        int AddRef() const override {return 0;}
        int Release() const override {return 0;}
    };

    
    class PeerConnectionObserverImp : public webrtc::PeerConnectionObserver {
    private:
        Conductor& parent;
    public:
        PeerConnectionObserverImp(Conductor& parent) : parent(parent) {}
        virtual ~PeerConnectionObserverImp () {}
        //PeerConnectionObserver implementation. 
        void OnSignalingChange(
            webrtc::PeerConnectionInterface::SignalingState new_state) override{
            fprintf(stdout, "PeerConnectionObserverImp::OnSignalingChange: %d\n", new_state);
        }
        
        // Called when a remote stream is added
        void OnAddStream(
            rtc::scoped_refptr<webrtc::MediaStreamInterface> stream) override {
        }
        virtual void OnAddStream(webrtc::MediaStreamInterface* stream) override {}
        
        void OnRemoveStream(
            rtc::scoped_refptr<webrtc::MediaStreamInterface> stream) override {
        }
        virtual void OnRemoveStream(webrtc::MediaStreamInterface* stream) override {}
        
        void OnDataChannel(rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel) override {
            parent.data_channel = data_channel;
        }
        void OnDataChannel(webrtc::DataChannelInterface* data_channel) override {}
        
        void OnRenegotiationNeeded() override {}
        void OnIceConnectionChange(
            webrtc::PeerConnectionInterface::IceConnectionState new_state) override{
            fprintf(stdout, "PeerConnectionObserverImp::OnIceConnectionChange:%d\n", new_state);
            if (new_state == webrtc::PeerConnectionInterface::IceConnectionState::kIceConnectionDisconnected) {
                if(parent.getVideoChannelStatus() || parent.getDataChannelStatus()) {
                    pushEvent(WEBRTC_EVENT_RENDER_LEAVE, parent.peerName, "");
                }
            }
        }
        void OnIceGatheringChange(webrtc::PeerConnectionInterface::IceGatheringState new_state) override{
        }
        
        void OnIceCandidate(const webrtc::IceCandidateInterface* candidate) override {
            Json::Value jmessage;
            Json::Value message_mid(candidate->sdp_mid());
            Json::Value message_mline_index(candidate->sdp_mline_index());
            jmessage[kCandidateSdpMidName] = message_mid;
            jmessage[kCandidateSdpMlineIndexName] = message_mline_index;
            std::string sdp;
            if (!candidate->ToString(&sdp)) {
                fprintf(stderr, "Failed to serialize candidate\n");
                return;
            }
            jmessage[kCandidateSdpName] = sdp;
            parent.SendMessage("candidate", jmessage);
        }
        void OnIceConnectionReceivingChange(bool receiving) override {
            fprintf(stdout, "PeerConnectionObserverImp::OnIceConnectionReceivingChange\n");
        }
    };
    
    
public:
    std::string peerName;
    SetSessionDescriptionObserverImp set_session_description_observer;
    CreateSessionDescriptionObserverImp create_session_description_observer;
    PeerConnectionObserverImp peer_connection_observer;
    DataChannelObserverImp data_channel_observer;
    rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection_;
    bool* hasPeerPointer = NULL;  // ponit to hasPeer{1 2 or 3}
    bool hasInitialize = false;
    cricket::VideoCapturer* video_capturer = nullptr;
    rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel = NULL;
    std::string filePath = "";
    int fileSize = 0;
    bool haveNeogotion = false;
    bool saveDataOver = false;
    bool fileTransformStop = false;
    bool fileTransformParameterAccepted = false;
    
    bool getVideoChannelStatus() {
        std::lock_guard<std::mutex> lock_channel(conductor_channel_used_webrtc_lock, std::adopt_lock);
        return videoChannelUsed;
    }
    bool getDataChannelStatus() {
        std::lock_guard<std::mutex> lock_channel(conductor_channel_used_webrtc_lock, std::adopt_lock);
        return dataChannelUsed;
    }
    void setVideoChannelStatus(bool status) {
        std::lock_guard<std::mutex> lock_channel(conductor_channel_used_webrtc_lock, std::adopt_lock);
        videoChannelUsed = status;
        if(!videoChannelUsed && !dataChannelUsed) {
            fprintf(stdout, "setVideoChannelStatus: conductor Close\n");
            fprintf(stdout, "setVideoChannelStatus: hasPeer pointer %p %p %p, hasPeerPointer: %p\n",
                    &hasPeerWebrtc1, &hasPeerWebrtc2, &hasPeerWebrtc3, hasPeerPointer);
            Close();
        }
    }
    void setDataChannelStatus(bool status) {
        std::lock_guard<std::mutex> lock_channel(conductor_channel_used_webrtc_lock, std::adopt_lock);
        dataChannelUsed = status;
        if(!videoChannelUsed && !dataChannelUsed) {
            fprintf(stdout, "setDataChannelStatus: conductor Close\n");
            fprintf(stdout, "setDataChannelStatus: hasPeer pointer %p %p %p, hasPeerPointer: %p\n",
                    &hasPeerWebrtc1, &hasPeerWebrtc2, &hasPeerWebrtc3, hasPeerPointer);
            Close();
        }
    }

    /* callback function to notify Conductor CreateAnswer or CreateOffer is success */
    void OnSuccessCallBack(webrtc::SessionDescriptionInterface* desc){
        std::string sdp;
        desc->ToString(&sdp);
        /*
        std::string find_str("a=setup:active");
        std::string replace_str("a=setup:passive");
        for(std::string::size_type i = 0; (i = sdp.find(find_str, i)) != std::string::npos;) {
            sdp.replace(i, find_str.length(), replace_str);
            i += replace_str.length();
        }
        webrtc::SdpParseError error;
        webrtc::SessionDescriptionInterface* session_description(
            webrtc::CreateSessionDescription(desc->type(), sdp, &error));
        */
        peer_connection_->SetLocalDescription(&set_session_description_observer, desc);
        Json::Value jmessage;
        Json::Value message_sdp(sdp);
        Json::Value message_type(desc->type());
        jmessage[kSessionDescriptionTypeName] = message_type;
        jmessage[kSessionDescriptionSdpName] = message_sdp;
        SendMessage(desc->type(),jmessage);
    }
    
    Conductor() : create_session_description_observer(*this), peer_connection_observer(*this),
                  data_channel_observer(this) {}

    void Close() {
        peerName = "";
        if(hasPeerPointer != NULL) {
            (*hasPeerPointer) = false;
        }
        hasPeerPointer = NULL;
        hasInitialize = false;
        video_capturer = nullptr;
        if(data_channel.get() != NULL) {
            data_channel->Close();
        }
        data_channel = NULL;
        filePath = "";
        fileSize = -1;
        haveNeogotion = false;
        saveDataOver = false;
        fileTransformStop = false;
        fileTransformParameterAccepted = false;
        if(getVideoChannelStatus()) {
            setVideoChannelStatus(false);
        }
        if(getDataChannelStatus()) {
            setDataChannelStatus(false);
        }
        DeletePeerConnection();
    }

    bool InitializePeerConnection() {
        if(peer_connection_.get() != NULL) {
            fprintf(stderr, "InitializePeerConnection ERROR: Initialized already\n");
            return false;
        }
        if (!CreatePeerConnection(true)) {
            fprintf(stderr, "Failed to initialize PeerConnectionFactory \n");
            DeletePeerConnection();
        }
        hasInitialize = true;
        AddStreams();
        webrtc::DataChannelInit config;
        data_channel = peer_connection_->CreateDataChannel("data_channel", &config);
        data_channel->RegisterObserver(&data_channel_observer);
        return peer_connection_.get() != NULL;
    }

    void SendMessage(const std::string& type, Json::Value& payload) {
        Json::Value payload_type(type);
        Json::Value payload_name(peerName);
        Json::Value message;
        message["type"] = payload_type;
        message["name"] = payload_name;
        message["payload"] = payload;
        std::string message_str = rtc::JsonValueToString(message);
        websocketSendMessage(message_str);
    }

    void handleOffer(Json::Value& jmessage) {
        std::string sdp_type;
        std::string sdp;
        rtc::GetStringFromJson(jmessage["type"], &sdp_type);
        rtc::GetStringFromJson(jmessage["sdp"], &sdp);
        webrtc::SdpParseError error;
        webrtc::SessionDescriptionInterface* session_description(
            webrtc::CreateSessionDescription(sdp_type, sdp, &error));
        if (!session_description) {
            fprintf(stderr, "Can't parse received session description message. SdpParseError was: %s\n",
                    error.description.c_str());
            return;
        }
        peer_connection_->SetRemoteDescription(
            &set_session_description_observer, session_description);
        peer_connection_->CreateAnswer(&create_session_description_observer, NULL);
        return;
    }

    int sendFileTransformparameter() {
        std::ifstream downloadFile(filePath);
        if (!downloadFile) {
            return WEBRTC_DOWNLOAD_FILE_NOT_EXIST;
        }
        downloadFile.seekg(0, std::ios::end);
        std::streampos fileSize_ = downloadFile.tellg();
        fileSize = int(fileSize_);
        downloadFile.close();

        if(fileSize > 0) {
            Json::Value jmessage;
            Json::Value messageFileName(filePath.substr(filePath.find_last_of("/\\") + 1));
            Json::Value messageFileSize(fileSize);
            jmessage["fileName"] = messageFileName;
            jmessage["fileSize"] = messageFileSize;
            SendMessage("file-transform-parameter", jmessage);
        } else {
            fprintf(stderr, "ERROR: fileSize == 0\n");
            Json::Value payload("");
            SendMessage("file-transform-file-size-zero", payload);
            return WEBRTC_DOWNLOAD_FILE_SIZE_ERROR;
        }
        return WEBRTC_SUCCESS;
    }
    
    int startFileTransform() {
        int ret = sendFileTransformparameter();
        if(ret != 0) {
            fprintf(stderr, "ERROR: sendFileTransformparameter failed\n");
            return ret;
        } else {
            if(!haveNeogotion) {
                handleOfferLoopback(peerName);
            } else {
                fprintf(stderr, "ERROR: haveNeogotion\n");
            }
            return WEBRTC_SUCCESS;
        }
    }
    
    void handleOfferLoopback(const std::string& peer_name) {
        webrtc::FakeConstraints constraints;
        constraints.AddMandatory(webrtc::MediaConstraintsInterface::kOfferToReceiveVideo, "false");
        constraints.AddMandatory(webrtc::MediaConstraintsInterface::kOfferToReceiveAudio, "false");
        peer_connection_->CreateOffer(&create_session_description_observer, &constraints);
        haveNeogotion = true;
    }

    void handleAnswer(Json::Value& jmessage) {
        std::string sdp_type;
        std::string sdp;
        rtc::GetStringFromJson(jmessage["type"], &sdp_type);
        rtc::GetStringFromJson(jmessage["sdp"], &sdp);
        webrtc::SdpParseError error;
        webrtc::SessionDescriptionInterface* session_description(
            webrtc::CreateSessionDescription(sdp_type, sdp, &error));
        if (!session_description) {
            fprintf(stderr, "Can't parse received session description message. SdpParseError was: %s\n",
                    error.description.c_str());
            return;
        }
        peer_connection_->SetRemoteDescription(
            &set_session_description_observer, session_description);
    
        pushEvent(WEBRTC_EVENT_RENDER_JOIN, peerName, "");
        return;
    }

    void handleCandidate(Json::Value& jmessage) {
        std::string sdp_mid;
        int sdp_mlineindex = 0;
        std::string sdp;
        if (!rtc::GetStringFromJsonObject(jmessage, kCandidateSdpMidName, &sdp_mid) ||
            !rtc::GetIntFromJsonObject(jmessage, kCandidateSdpMlineIndexName, &sdp_mlineindex) ||
            !rtc::GetStringFromJsonObject(jmessage, kCandidateSdpName, &sdp)) {
            return;
        }
        webrtc::SdpParseError error;
        std::unique_ptr<webrtc::IceCandidateInterface> candidate(
            webrtc::CreateIceCandidate(sdp_mid, sdp_mlineindex, sdp, &error));
        if (!candidate.get()) {
            fprintf(stderr, "Can't parse received session description message. SdpParseError was: %s\n",
                    error.description.c_str());
            return;
        }
        if (!peer_connection_->AddIceCandidate(candidate.get())) {
            return;
        }
        return;
    }

    static void* sendData(void* conductor_) {
        fprintf(stdout, "sendData starting\n");
        Conductor* conductor = (Conductor *)conductor_;
        int timeout = 0;
        while(!conductor->saveDataOver && !conductor->fileTransformStop && !conductor->fileTransformParameterAccepted) {
            //wait 30 seconds for peer accept fileTransformParameter
            std::this_thread::sleep_for(std::chrono::microseconds(500000));
            timeout += 1;
            if (timeout >= 60) {
                conductor->setDataChannelStatus(false);
                fprintf(stderr, "file transform parameter have not accepted, timeout, quit\n");
                pthread_exit(NULL);
            } else {
                fprintf(stdout, "file transform parameter have not accepted, sleeping\n");
            }
        }
        
        fprintf(stdout, "sendData start\n");        
        std::ifstream downloadFile(conductor->filePath, std::ios::binary);
        int chunkSize = 1024 * 16;
        char chunk[chunkSize] = {0};
        int fileSizeCurrent = 0;
        int readSize = 0;
        int chunkNum = 0;
        fprintf(stdout, "sendData start: %d\n", conductor->fileSize);
        while(fileSizeCurrent < conductor->fileSize && !conductor->fileTransformStop) {
            if(fileSizeCurrent + chunkSize <= conductor->fileSize) {
                readSize = chunkSize;
            } else {
                readSize = conductor->fileSize - fileSizeCurrent;
            }
            downloadFile.read(chunk, readSize);
            webrtc::DataBuffer buffer(rtc::CopyOnWriteBuffer(chunk, readSize), true);
            if(conductor->data_channel.get() == NULL) {
                conductor->saveDataOver = true;
                break;
            }
            conductor->data_channel->Send(buffer);
            fileSizeCurrent += readSize;
            chunkNum += 1;
            if((chunkNum % 5 == 0) && conductor->getVideoChannelStatus()) {
                std::this_thread::sleep_for(std::chrono::microseconds(1000000));
            }
            fprintf(stdout, "sendData start fileSizeCurrent: %d\n", fileSizeCurrent);
        }
        downloadFile.close();
        while(!conductor->saveDataOver && !conductor->fileTransformStop) {
            std::this_thread::sleep_for(std::chrono::microseconds(500000));
        }
        fprintf(stdout, "sendData over\n");
        conductor->setDataChannelStatus(false);
        pthread_exit(NULL);
    }

    void notifyPeerClose() {
        if(hasPeerPointer != NULL && *hasPeerPointer) {
            fprintf(stdout, "notifyPeerClose\n");
            Json::Value jmessage("");
            SendMessage("leave", jmessage);
        }
        Close();
    }
    
private:
    bool videoChannelUsed = false;
    bool dataChannelUsed = false;
    std::map<std::string, rtc::scoped_refptr<webrtc::MediaStreamInterface> > active_streams_;

    std::string get_system_timestamp(long time_to_live) {
        std::time_t result = std::time(nullptr);
        char timestamp_char[16];
        sprintf(timestamp_char, "%ld", result + time_to_live);
        return std::string(timestamp_char);
    }
    
    std::string GetServerTimestamp() {
        long time_to_live = 600;
        /* lookup the ip address */
        struct hostent* server_ip = gethostbyname(serverAddressInternalWebrtc.c_str());
        if (server_ip == NULL) {
            return get_system_timestamp(time_to_live);
        }
        
        int sock = socket(AF_INET, SOCK_STREAM, 0);
        struct sockaddr_in serv_addr;
        memset(&serv_addr, 0, sizeof(serv_addr));
        serv_addr.sin_family = AF_INET;
        memcpy(&serv_addr.sin_addr.s_addr, server_ip->h_addr, server_ip->h_length);
        serv_addr.sin_port = htons(3033);  //端口
        if(sock < 0 || connect(sock, (struct sockaddr*)&serv_addr, sizeof(serv_addr)) < 0) {
            fprintf(stderr, "GET /timestamp connect failed\n");
            return get_system_timestamp(time_to_live);
        }

        const int buffer_size = 256;
        char buffer_send[buffer_size];
        char buffer_receive[buffer_size];
        for(int i = 0; i < buffer_size; i++) {
            buffer_send[i] = '\0';
            buffer_receive[i] = '\0';
        }
        sprintf(buffer_send, "GET /timestamp HTTP/1.0\r\n\r\n");
        write(sock, buffer_send, sizeof(buffer_send)-1);
        read(sock, buffer_receive, sizeof(buffer_receive)-1);
        std::string buffer_receive_str(buffer_receive);
        std::string mark_string = "timestamp_is_";
        std::size_t mark_location = buffer_receive_str.find(mark_string);
        char timestamp[11];
        for(int i = 0; i < 10; i++){
            timestamp[i] = buffer_receive[mark_location + mark_string.size() + i];
        }
        timestamp[10] = '\0';
        std::string timestamp_str(timestamp);
        close(sock);
        fprintf(stdout, "%s\n", timestamp_str.c_str());
        return timestamp_str;
    }
    
    bool CreatePeerConnection(bool dtls) {
        char turn_key[] = "t=u408021891333";
        std::string coturnUserName = GetServerTimestamp();
        coturnUserName = coturnUserName + ":" + userNameInternalWebrtc;
        unsigned char digest[EVP_MAX_MD_SIZE] = {'\0'};
        unsigned int digest_len = 0;
        HMAC(EVP_sha1(), turn_key, strlen(turn_key), (unsigned char*)coturnUserName.c_str(),
             coturnUserName.size(), digest, &digest_len);
        std::string password;
        rtc::Base64::EncodeFromArray(digest, digest_len, &password);
        std::string server_url = "turn:" + host_address_webrtc;
        std::string server_urls[] = {
            server_url + ":3478?transport=udp",
            server_url + ":3478?transport=tcp",
            server_url + ":3479?transport=udp",
            server_url + ":3479?transport=tcp",
        };
        webrtc::PeerConnectionInterface::RTCConfiguration config;
        fprintf(stdout, "%s %s\n", coturnUserName.c_str(), password.c_str());
        for(int i = 0; i < 4; i++){
            webrtc::PeerConnectionInterface::IceServer server;
            server.uri = server_urls[i];
            server.username = coturnUserName;
            server.password = password;
            config.servers.push_back(server);
        }
        
        webrtc::FakeConstraints constraints;
        if (dtls) {
            constraints.AddOptional(webrtc::MediaConstraintsInterface::kEnableDtlsSrtp,
                                    "true");
        } else {
            constraints.AddOptional(webrtc::MediaConstraintsInterface::kEnableDtlsSrtp,
                                    "false");
        }

        constraints.AddMandatory(webrtc::MediaConstraintsInterface::kOfferToReceiveVideo, "false");
        constraints.AddMandatory(webrtc::MediaConstraintsInterface::kOfferToReceiveAudio, "false");
        peer_connection_ = peer_connection_factory_webrtc->CreatePeerConnection(
            config, &constraints, NULL, NULL, &peer_connection_observer);
        return peer_connection_.get() != NULL;
    }

    void DeletePeerConnection() {
        peer_connection_ = NULL;
        if (active_streams_.find(kStreamLabel) != active_streams_.end()) {
            active_streams_.clear();
        }
        //peer_connection_factory_webrtc = NULL;
    }

    void AddStreams() {
        if (active_streams_.find(kStreamLabel) != active_streams_.end())
            return;  // Already added.
        /*
        rtc::scoped_refptr<webrtc::AudioTrackInterface> audio_track(
            peer_connection_factory_webrtc->CreateAudioTrack(
                kAudioLabel, peer_connection_factory_webrtc->CreateAudioSource(NULL)));
        */
        webrtc::FakeConstraints constraints;
        constraints.AddMandatory(webrtc::MediaConstraintsInterface::kMaxFrameRate, 15);
        rtc::scoped_refptr<webrtc::VideoTrackInterface> video_track(
            peer_connection_factory_webrtc->CreateVideoTrack(
                kVideoLabel,
                peer_connection_factory_webrtc->CreateVideoSource(OpenVideoCaptureDevice(),
                                                            &constraints)));
        
        rtc::scoped_refptr<webrtc::MediaStreamInterface> stream =
            peer_connection_factory_webrtc->CreateLocalMediaStream(kStreamLabel);
        //stream->AddTrack(audio_track);
        stream->AddTrack(video_track);
        if (!peer_connection_->AddStream(stream)) {
            fprintf(stderr, "Adding stream to PeerConnection failed\n");
        }
        typedef std::pair<std::string,
                          rtc::scoped_refptr<webrtc::MediaStreamInterface> >
            MediaStreamPair;
        active_streams_.insert(MediaStreamPair(stream->label(), stream));
    }
        
    cricket::VideoCapturer* OpenVideoCaptureDevice(){
        std::vector<std::string> device_names;
        {
            std::unique_ptr<webrtc::VideoCaptureModule::DeviceInfo> info(
                webrtc::VideoCaptureFactory::CreateDeviceInfo(0));
            if (!info) {
                return nullptr;
            }
            int num_devices = info->NumberOfDevices();
            for (int i = 0; i < num_devices; ++i) {
                const uint32_t kSize = 256;
                char name[kSize] = {0};
                char id[kSize] = {0};
                if (info->GetDeviceName(i, name, kSize, id, kSize) != -1) {
                    device_names.push_back(name);
                }
            }
        }

        cricket::WebRtcVideoDeviceCapturerFactory factory;
        cricket::VideoCapturer* capturer = nullptr;
        for (const auto& name : device_names) {
            capturer = factory.Create(cricket::Device(name, 0));
            if (capturer) {
                break;
            }
        }
        video_capturer = capturer;
        return capturer;
    }
};


Conductor conductor1;
Conductor conductor2;
Conductor conductor3;
rtc::Thread* thread_rtc = NULL;

void* sendPing(void* args) {
    while(!stopedWebrtc) {
        if(loginInSuccessWebrtc) {
            Json::Value message;
            Json::Value payload("");
            Json::Value payload_type("ping-signal");
            message["type"] = payload_type;
            message["payload"] = payload;
            std::string message_str = rtc::JsonValueToString(message);
            websocketSendMessage(message_str);
        }
        std::this_thread::sleep_for(std::chrono::microseconds(10000000));
    }
    pthread_exit(NULL);
}

void handleLogin(Json::Value jmessage) {
    bool login_success = false;
    rtc::GetBoolFromJson(jmessage["success"], &login_success);
    if(!loginInSuccessWebrtc) {
        if(login_success){
            loginInSuccessWebrtc = true;
            pushEvent(WEBRTC_EVENT_SERVER_LOGIN, "", "");
        } else {
            fprintf(stderr, "handleLogin: device name invaild\n");
            srand((unsigned)time(NULL));
            userNameInternalWebrtc = userNameInternalWebrtc + "-" + std::to_string(rand());
            fprintf(stdout, "handleLogin: create new device name %s\n", userNameInternalWebrtc.c_str());
            Json::Value payload_type("login");
            Json::Value payload_name(userNameInternalWebrtc);
            Json::Value payload;
            payload["type"] = payload_type;
            payload["name"] = payload_name;
            std::string payload_str = rtc::JsonValueToString(payload);
            websocketSendMessage(payload_str);
        }
    } else {
        fprintf(stdout, "handleLogin: loginInSuccessWebrtc already\n");
    }
}

Conductor* findConductor(const std::string& peerName) {
    if(conductor1.peerName == peerName) {
        return &conductor1;
    } else if(conductor2.peerName == peerName) {
        return &conductor2;
    } else if(conductor3.peerName == peerName) {
        return &conductor3;
    } else {
        return nullptr;
    }
}

void handle_message(const std::string& message) {
    fprintf(stdout, "got message%s\n", message.c_str());
    Json::Reader reader;
    Json::Value jmessage;
    if (!reader.parse(message, jmessage)) {
        return;
    }
    std::string message_type;
    rtc::GetStringFromJson(jmessage["type"], &message_type);
    std::string message_from;
    rtc::GetStringFromJson(jmessage["from"], &message_from);
    Conductor* conductor = nullptr;
    if (message_from != "") {
        conductor = findConductor(message_from);
        if(conductor != nullptr){
            fprintf(stdout, "handle_message: hasPeer %d %d %d, video_capturer: %p, hasInitialize %d\n",
                    hasPeerWebrtc1, hasPeerWebrtc2, hasPeerWebrtc3, conductor->video_capturer, conductor->hasInitialize);
            fprintf(stdout, "handle_message: videoChannelUsed: %d, dataChannelUsed: %d\n",
                    conductor->getVideoChannelStatus(), conductor->getDataChannelStatus());
            fprintf(stdout, "handle_message: hasPeer pointer %p %p %p, conductor->hasPeerPointer: %p\n",
                    &hasPeerWebrtc1, &hasPeerWebrtc2, &hasPeerWebrtc3, conductor->hasPeerPointer);
        } else if(message_type != "message" && message_type != "file-transform" && message_type != "offer-loopback") {
            fprintf(stdout, "peer has closed\n");
            return;
        } else if (!hasPeerWebrtc1) {
            hasPeerWebrtc1 = true;
            conductor = &conductor1;
            conductor->hasPeerPointer = &hasPeerWebrtc1;
            fprintf(stdout, "handle_message_1\n");
        } else if (!hasPeerWebrtc2) {
            hasPeerWebrtc2 = true;
            conductor = &conductor2;
            conductor->hasPeerPointer = &hasPeerWebrtc2;
            fprintf(stdout, "handle_message_2\n");
        } else if (!hasPeerWebrtc3) {
            hasPeerWebrtc3 = true;
            conductor = &conductor3;
            conductor->hasPeerPointer = &hasPeerWebrtc3;
            fprintf(stdout, "handle_message_3\n");
        } else {
            fprintf(stdout, "ERROR: total number of peer exceeding 3\n");
            Json::Value message;
            Json::Value message_type("peer-excced-3");
            Json::Value message_name(message_from);
            Json::Value payload("");
            message["type"] = message_type;
            message["name"] = message_name;
            message["payload"] = payload;
            websocketSendMessage(rtc::JsonValueToString(message));
            return;
        }
        if (conductor->peerName != message_from) {
            conductor->peerName = message_from;
        }
    }
    
    if(message_type == "message") {
        std::string message_payload = rtc::JsonValueToString(jmessage["payload"]);
        std::string message_payload_str = "";
        for(std::string::const_iterator it = message_payload.begin(); it != message_payload.end(); ++it) {
            if (*it != '"') {
                message_payload_str += *it;
            }
        }
        pushEvent(WEBRTC_EVENT_MESSAGE, message_from, message_payload_str);
    } else if(message_type == "save-data-over") {
        conductor->saveDataOver = true;
        conductor->setDataChannelStatus(false);
    } else if(message_type == "login") {
        handleLogin(jmessage);
    } else if(message_type == "file-transform-parameter-accept") {
        conductor->fileTransformParameterAccepted = true;
    } else if(message_type == "file-transform-cancel") {
        pushEvent(WEBRTC_EVENT_FILE_ABORT, message_from, "");
    } else if(message_type == "file-transform") {
        std::string fileName_;
        rtc::GetStringFromJson(jmessage["payload"]["fileName"], &fileName_);
        pushEvent(WEBRTC_EVENT_FILE_PULL_REQUEST, conductor->peerName, "fileName=" + fileName_);
    } else if(message_type == "offer-loopback") {
        conductor->handleOfferLoopback(message_from);
    } else if(message_type == "offer") {
        conductor->handleOffer(jmessage["payload"]);
    } else if(message_type == "answer") {
        conductor->handleAnswer(jmessage["payload"]);
    } else if(message_type == "candidate") {
        conductor->handleCandidate(jmessage["payload"]);
    } else if(message_type == "leave") {
        pushEvent(WEBRTC_EVENT_RENDER_LEAVE, conductor->peerName, "");
    } else if(message_type == "server_kick_out") {
        pushEvent(WEBRTC_EVENT_SERVER_KICK_OUT, "", "");
    } else if(message_type == "pong-signal") {
        ;
    } else {
        ;
    }
}


void thread_entry() {
    std::stringstream thread_id_str;
    thread_id_str << std::this_thread::get_id();
    fprintf(stdout, "thread_entry: %s\n", thread_id_str.str().c_str());
    
    if(thread_rtc == NULL && peer_connection_factory_webrtc == NULL) {
        peer_connection_factory_webrtc = webrtc::CreatePeerConnectionFactory();
        if (peer_connection_factory_webrtc.get() == nullptr) {
            fprintf(stderr, "Error on CreatePeerConnectionFactory.");
            return;
        }
        rtc::PhysicalSocketServer socket_server;
        thread_rtc = rtc::Thread::Current();
        thread_rtc->set_socketserver(&socket_server);
        thread_rtc->Run();
        thread_rtc->set_socketserver(nullptr);
    }

    thread_id_str.str("");
    thread_id_str << std::this_thread::get_id();
    fprintf(stdout, "thread_entry over: %s\n", thread_id_str.str().c_str());
}

void web_SocketClient() {
    std::string host = serverAddressInternalWebrtc;
    std::string port = "3000";
    host_address_webrtc = host;
    rtc::LogMessage::SetLogToStderr(true);
    std::stringstream thread_id_str;
    thread_id_str << std::this_thread::get_id();
    fprintf(stdout, "web_SocketClient thread: %s\n", thread_id_str.str().c_str());
    {
        std::lock_guard<std::mutex> lock_ws(websocket_client_webrtc_lock, std::adopt_lock);
        websocket_client_webrtc = easywsclient::WebSocket::from_url("ws://" + host + ":" + port);
        if(websocket_client_webrtc) {
            fprintf(stdout, "WebSocket created success\n");
        } else {
            pushEvent(WEBRTC_EVENT_CONNECT_CLOSED, "", "");
            fprintf(stdout, "WebSocket create failed\n");
            return;
        }
    }
    
    rtc::InitializeSSL();
    std::thread th(thread_entry);
    while(peer_connection_factory_webrtc.get() == nullptr) {
        std::this_thread::sleep_for(std::chrono::microseconds(100000));
    }
    easywsclient::WebSocket::readyStateValues webSocketState;
    {
        std::lock_guard<std::mutex> lock_ws(websocket_client_webrtc_lock, std::adopt_lock);
        webSocketState = websocket_client_webrtc->getReadyState();
        pthread_t pthread_id;
        pthread_create(&pthread_id, NULL, sendPing, NULL);
        pthread_detach(pthread_id);
    }

    if(!loginInSuccessWebrtc) {
        Json::Value payload_type("login");
        Json::Value payload_name(userNameInternalWebrtc);
        Json::Value payload;
        payload["type"] = payload_type;
        payload["name"] = payload_name;
        std::string payload_str = rtc::JsonValueToString(payload);
        websocketSendMessage(payload_str);
    }
    while (webSocketState == easywsclient::WebSocket::OPEN) {
        if(stopedWebrtc){
            break;
        }
        {
            std::lock_guard<std::mutex> lock_ws(websocket_client_webrtc_lock, std::adopt_lock);
            websocket_client_webrtc->poll();
            websocket_client_webrtc->dispatch(handle_message);
        }

        std::this_thread::sleep_for(std::chrono::microseconds(400000));
        {
            std::lock_guard<std::mutex> lock_ws(websocket_client_webrtc_lock, std::adopt_lock);
            webSocketState = websocket_client_webrtc->getReadyState();
            if(webSocketState != easywsclient::WebSocket::OPEN) {
                fprintf(stderr, "webSocketState != easywsclient::WebSocket::OPEN\n");
                pushEvent(WEBRTC_EVENT_CONNECT_CLOSED, "", "");
            } else {
                ; //fprintf(stdout, "web_SocketClient: webSocketState: OPEN SUCCESS\n");
            }
        }
    }
    fprintf(stderr, "web_SocketClient: ws loop over\n");
    delete websocket_client_webrtc;
    websocket_client_webrtc = NULL;
    th.join();
    rtc::CleanupSSL();
    //th.detach();
    fprintf(stderr, "web_SocketClient: over\n");
}


int initializeWebrtc(const std::string& deviceName, const std::string& serverAddress, std::string& realDeviceName) {
    fprintf(stdout, "initializeWebrtc\n");
    {
        std::lock_guard<std::mutex> lock_stoped_webrtc(stopedWebrtcLock, std::adopt_lock);
        if(!stopedWebrtc) {
            fprintf(stdout, "WEBRTC_INIT_ALREADY\n");
            return WEBRTC_INIT_ALREADY;
        }
        stopedWebrtc = false;
    }
    std::stringstream thread_id_str;
    thread_id_str << std::this_thread::get_id();
    fprintf(stdout, "initializeWebrtc thread: %s\n", thread_id_str.str().c_str());
    //srand((unsigned)time(NULL));
    userNameInternalWebrtc = deviceName; // + "-" + std::to_string(rand());
    serverAddressInternalWebrtc = serverAddress;
    std::thread thread_websocket(web_SocketClient);
    thread_websocket.detach();
    realDeviceName = userNameInternalWebrtc;
    fprintf(stdout, "initializeWebrtc: deviceName: %s, serverAddress: %s", deviceName.c_str(), serverAddress.c_str());
    return WEBRTC_SUCCESS;
}

void startVideoWebrtc(const std::string& peerName) {
    fprintf(stdout, "startVideoWebrtc peerName: %s\n", peerName.c_str());
    if(!loginInSuccessWebrtc) {
        return;
    }
    Conductor* conductor = findConductor(peerName);
    if(conductor == nullptr) {
        return;
    } else {
        conductor->setVideoChannelStatus(true);
        if(!conductor->hasInitialize) {
            conductor->InitializePeerConnection();
        } else {
            fprintf(stdout, "startVideoWebrtc: hasInitialize");
        }
        conductor->handleOfferLoopback(peerName);
    }
}

void stopVideoWebrtc(const std::string& peerName) {
    Conductor* conductor = findConductor(peerName);
    if(conductor == nullptr) {
        return;
    } else {
        conductor->setVideoChannelStatus(false);
    }
}

void stopVideoAndNotifyPeerWebrtc(const std::string& peerName) {
    Conductor* conductor = findConductor(peerName);
    if(conductor == nullptr) {
        return;
    } else {
        if(conductor->hasPeerPointer != NULL && *(conductor->hasPeerPointer)) {
            fprintf(stdout, "notifyPeerClose\n");
            Json::Value jmessage("");
            conductor->SendMessage("leave", jmessage);
        }
        conductor->setVideoChannelStatus(false);
    }
}

int notifyWebrtc(const std::string& message) {
    if (!hasPeerWebrtc1 && !hasPeerWebrtc2 && !hasPeerWebrtc3) {
        return WEBRTC_PEER_NOT_EXIST;
    } else {
        Json::Value payload(message);
        if(hasPeerWebrtc1) {
            conductor1.SendMessage("message", payload);
        }
        if(hasPeerWebrtc2) {
            conductor2.SendMessage("message", payload);
        }
        if(hasPeerWebrtc3) {
            conductor3.SendMessage("message", payload);
        }
        return WEBRTC_SUCCESS;
    }
}

int sendMessageWebrtc(const std::string& peerName, const std::string& message) {
    Conductor* conductor = findConductor(peerName);
    if(conductor != nullptr) {
        Json::Value payload(message);
        conductor->SendMessage("message", payload);
        return WEBRTC_SUCCESS;
    } else {
        return WEBRTC_PEER_NOT_EXIST;
    }
}


void sendFrameToWebrtc(uint8_t* videoFrame, size_t videoFrameLength, int width, int height, bool IsKeyFrame) {
    if(stopedWebrtc || !loginInSuccessWebrtc) {
        return;
    }
    webrtc::VideoCaptureCapability frameInfo;
    frameInfo.width = width;
    frameInfo.height = height;
    frameInfo.rawType = webrtc::kVideoNV21;
    frameInfo.codecType = webrtc::kVideoCodecH264;
    frameInfo.keyframe = IsKeyFrame;
    if(hasPeerWebrtc1) {
        if(conductor1.getVideoChannelStatus() && conductor1.video_capturer != nullptr) {
            ((webrtc::videocapturemodule::VideoCaptureImpl *)
             (((cricket::WebRtcVideoCapturer *)conductor1.video_capturer)->get_module_()).get())->IncomingFrame(
                 videoFrame, videoFrameLength, frameInfo);
        }
    }
    if(hasPeerWebrtc2) {
        if(conductor2.getVideoChannelStatus() && conductor2.video_capturer != nullptr) {
            ((webrtc::videocapturemodule::VideoCaptureImpl *)
             (((cricket::WebRtcVideoCapturer *)conductor2.video_capturer)->get_module_()).get())->IncomingFrame(
                 videoFrame, videoFrameLength, frameInfo);
        }
    }
    if(hasPeerWebrtc3) {
        if(conductor3.getVideoChannelStatus() && conductor3.video_capturer != nullptr) {
            ((webrtc::videocapturemodule::VideoCaptureImpl *)
             (((cricket::WebRtcVideoCapturer *)conductor3.video_capturer)->get_module_()).get())->IncomingFrame(
                 videoFrame, videoFrameLength, frameInfo);
        }
    }
}

int fileAcceptWebrtc(const std::string& peerName, const std::string& filePath) {
    fprintf(stdout, "fileAcceptWebrtc%s\n", filePath.c_str());
    Conductor* conductor = findConductor(peerName);
    if(conductor == nullptr) {
        return WEBRTC_PEER_NOT_EXIST;
    }
    if(!conductor->hasInitialize) {
        conductor->InitializePeerConnection();
    }
    conductor->filePath = filePath;
    conductor->fileTransformStop = false;
    conductor->fileTransformParameterAccepted = false;
    conductor->saveDataOver = false;
    conductor->setDataChannelStatus(true);
    if(conductor->data_channel.get() != NULL && conductor->haveNeogotion) {
        int ret = conductor->sendFileTransformparameter();
        if(ret != 0) {
            return ret;
        } else {
            pthread_t pthread_id;
            pthread_create(&pthread_id, NULL, Conductor::sendData, (void* )(conductor));
            pthread_detach(pthread_id);
            return WEBRTC_SUCCESS;
        }
    } else {
        return conductor->startFileTransform();
    }
}

void fileRejectWebrtc(const std::string& peerName) {}

void fileCancelWebrtc(const std::string& peerName) {
    fprintf(stdout, "fileCancelWebrtc%s\n", peerName.c_str());
    Conductor* conductor = findConductor(peerName);
    if(conductor == nullptr) {
        return;
    }
    conductor->fileTransformStop = true;
    conductor->fileTransformParameterAccepted = true;
    conductor->saveDataOver = true;
    conductor->setDataChannelStatus(false);
}

int cleanupWebrtc() {
    fprintf(stdout, "cleanupWebrtc\n");
    {
        std::lock_guard<std::mutex> lock_stoped_webrtc(stopedWebrtcLock, std::adopt_lock);
        if(stopedWebrtc) {
            fprintf(stdout, "cleanupWebrtc: WEBRTC_UNINIT\n");
            return WEBRTC_UNINIT;
        }
        stopedWebrtc = true;
        loginInSuccessWebrtc = false;
    }
    
    if(hasPeerWebrtc1) {
        conductor1.notifyPeerClose();
    }
    if(hasPeerWebrtc2) {
        conductor2.notifyPeerClose();
    }
    if(hasPeerWebrtc3) {
        conductor3.notifyPeerClose();
    }
    /*
    peer_connection_factory_webrtc = NULL;
    if(thread_rtc != NULL) {
        fprintf(stdout, "cleanupWebrtc: quit thread_rtc\n");
        thread_rtc->Stop();
        thread_rtc->Quit();
        thread_rtc = NULL;
    } else {
        fprintf(stdout, "cleanupWebrtc: thread_rtc is NULL\n");
        } */
    fprintf(stdout, "cleanupWebrtc over\n");
    return WEBRTC_SUCCESS;
}

int getEventWebrtc(int& eventType, std::string& peerName, std::string& eventData, int timeout) {
    WebrtcEvent e;
    int timeout_;
    if(popEvent(e) == 0) {
        eventType = e.eventType;
        peerName = e.peerName;
        eventData = e.eventData;
        return WEBRTC_SUCCESS;
    } else {
        if (timeout == 0) {
            timeout_ = 20;
        } else {
            timeout_ = timeout;
        }
        do {
            std::this_thread::sleep_for(std::chrono::milliseconds(timeout_));
            if(popEvent(e) == 0) {
                eventType = e.eventType;
                peerName = e.peerName;
                eventData = e.eventData;
                return WEBRTC_SUCCESS;
            }
        } while (timeout == 0);
    }
    eventType = 0; // WEBRTC_EVENT_NULL
    return WEBRTC_TIMEOUT;
}

int main(int argc, char **argv){
}
